package com.wesley;

import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.Arrays;
import java.util.List;

/**
 * HBase操作类
 * @author:wesley
 */
public class HBaseUtil {

    /**
     *
     * 创建表
     * @param tableName 表明
     * @param cfs 列族
     * @return 是否成功
     */
    public static boolean createTable(String tableName,String[] cfs) {
        //对表操作需要admin实例
        try (HBaseAdmin admin = (HBaseAdmin) HBaseConn.getHBaseConnect().getAdmin()){
            if (admin.tableExists(TableName.valueOf(tableName))) {
                return false;
            }

            final HTableDescriptor descriptor = new HTableDescriptor(TableName.valueOf(tableName));
            Arrays.stream(cfs).forEach(cf->{
                HColumnDescriptor columnDescriptor =new HColumnDescriptor(cf);
                columnDescriptor.setMaxVersions(1);
                descriptor.addFamily(columnDescriptor);
            });
            admin.createTable(descriptor);
        }catch (Exception e) {
            e.printStackTrace();
        }
        return true;
    }


    /**
     *
     * 插入数据
     *
     * @param tableName 表名
     * @param rowKey 表示
     * @param cfName 列族
     * @param qualifier 列标识
     * @param data 数据
     * @return 是否插入成功
     */
    public static boolean putRow(String tableName, String rowKey,String cfName,String qualifier,String data){
        try (Table table = HBaseConn.getTable(tableName)){
            Put put = new Put(Bytes.toBytes(rowKey));
            put.addColumn(Bytes.toBytes(cfName), Bytes.toBytes(qualifier), Bytes.toBytes(data));
            table.put(put);
        }catch (Exception e) {
            e.printStackTrace();
        }
        return true;
    }


    /**
     * 批量插入数据
     *
     * @param tableName
     * @param puts
     * @return
     */
    public static boolean putRows(String tableName, List<Put> puts){
        try (Table table = HBaseConn.getTable(tableName)){
           table.put(puts);
        }catch (Exception e) {
            e.printStackTrace();
        }
        return true;
    }

    /**
     * 查出一行
     *
     * @param tableName 表明
     * @param rowKey 标识
     * @return
     */
    public static Result getRow(String tableName,String rowKey){
        try (Table table = HBaseConn.getTable(tableName)){
            Get get = new Get(Bytes.toBytes(rowKey));
            return table.get(get);
        }catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }


    /**
     *
     * 通过过滤器查询row
     *
     * @param tableName
     * @param rowKey
     * @param filterList 过滤器
     * @return
     */
    public static Result getRow(String tableName, String rowKey, FilterList filterList) {
        try (Table table = HBaseConn.getTable(tableName)){
            Get get = new Get(Bytes.toBytes(rowKey));
            get.setFilter(filterList);
            table.get(get);
        }catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    /**
     * 通过scan扫描拿数据
     *
     * @param tableName
     * @return
     */
    public static ResultScanner getScanner(String tableName) {
        try (Table table = HBaseConn.getTable(tableName)){
            Scan scan = new Scan();
            scan.setCaching(1000);
            return table.getScanner(scan);
        }catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    /**
     * 批量搜索数据
     * @param tableName 表名
     * @param startkey 开始key
     * @param endKey 截止key
     * @return
     */
    public static ResultScanner getScanner(String tableName,String startkey,String endKey) {
        try (Table table = HBaseConn.getTable(tableName)){
            Scan scan = new Scan();
            scan.setStartRow(Bytes.toBytes(startkey));
            scan.setStopRow(Bytes.toBytes(endKey));
            scan.setCaching(2000);
            return table.getScanner(scan);
        }catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    public static ResultScanner getScanner(String tableName,String startkey,String endKey,FilterList filterList) {
        try (Table table = HBaseConn.getTable(tableName)){
            Scan scan = new Scan();
            scan.setStartRow(Bytes.toBytes(startkey));
            scan.setStopRow(Bytes.toBytes(endKey));
            scan.setFilter(filterList);
            scan.setCaching(1000);
            return table.getScanner(scan);
        }catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }


    /**
     *
     * 删除一行
     *
     * @param tableName 表名
     * @param rowKey 标识
     * @return
     */
    public static boolean deteleRow(String tableName,String rowKey){
        try (Table table = HBaseConn.getTable(tableName)){
            Delete delete = new Delete(Bytes.toBytes(rowKey));
            table.delete(delete);
        }catch (Exception e) {
            e.printStackTrace();
        }
        return true;
    }

    /**
     * 删除列族
     * @param tableName
     * @param cfName
     * @return
     */
    public static boolean deleteColumnFamily(String tableName,String cfName) {
        try (HBaseAdmin admin = (HBaseAdmin) HBaseConn.getHBaseConnect().getAdmin()){
            admin.deleteColumn(tableName, cfName);
        }catch (Exception e) {
            e.printStackTrace();
        }
        return true;
    }

    /**
     * 删除列标识
     * @param tableName
     * @param rowKey
     * @param cfName
     * @param qualifier
     * @return
     */
    public static boolean deleteQualifier(String tableName,String rowKey,String cfName,String qualifier) {
        try (Table table = HBaseConn.getTable(tableName)){
            Delete delete = new Delete(Bytes.toBytes(rowKey));
            delete.addColumn(Bytes.toBytes(cfName), Bytes.toBytes(qualifier));
            table.delete(delete);
        }catch (Exception e) {
            e.printStackTrace();
        }
        return true;
    }

    /**
     * 删除表，在删除之前必须是disable
     * @param tableName
     * @throws IOException
     */
    public static void deleteTable(String tableName) throws IOException {
        HBaseAdmin admin = (HBaseAdmin) HBaseConn.getHBaseConnect().getAdmin();
        admin.disableTable(Bytes.toBytes(tableName));
        admin.deleteTable(Bytes.toBytes(tableName));
    }

}
